package com.fjwt.gz.components.mq.websocket.server;

import cn.hutool.json.JSONUtil;
import com.fjwt.gz.components.mq.constant.WebSocketConstants;
import com.fjwt.gz.components.mq.websocket.eventEntity.WebSocketFansEvent;
import com.fjwt.gz.components.mq.websocket.eventEntity.WebSocketImportEvent;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicInteger;

@ServerEndpoint("/websocket/{fanId}/{appId}")
@Component
@Slf4j
public class MinProgramWsChannelServer {

    private final static Logger logger = LoggerFactory.getLogger(MinProgramWsChannelServer.class);

    // 使用AtomicInteger保证线程安全
    private static final AtomicInteger onlineClientSize = new AtomicInteger(0);

    // 使用ConcurrentHashMap保证线程安全
    private static final Map<String, Set<MinProgramWsChannelServer>> wsUserIdMap = new ConcurrentHashMap<>();

    private Session session;
    private String fanId;
    private String appId;

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("fanId") String fanId, @PathParam("appId") String appId) {
        try {
            this.fanId = fanId;
            this.appId = appId;
            this.session = session;

            // 设置空闲超时时间（毫秒）
//            session.setMaxIdleTimeout(300000); // 5分钟

            wsUserIdMap.computeIfAbsent(fanId, k -> new CopyOnWriteArraySet<>()).add(this);

            int count = onlineClientSize.incrementAndGet();
            logger.info("fanId[{}], appId[{}] 连接开启监听！当前在线人数为{}", fanId, appId, count);
        } catch (Exception e) {
            logger.error("ws监听异常fanId[{}],appId[{}]", fanId, appId, e);
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        wsUserIdMap.computeIfPresent(this.fanId, (k, v) -> {
            v.remove(this);
            return v.isEmpty() ? null : v;
        });

        int count = onlineClientSize.decrementAndGet();
        logger.info("fanId[{}],fanId[{}],连接关闭！当前在线人数为{}", fanId, appId, count);
    }

    /**
     * 接收客户端消息的核心方法
     * @param message 客户端发送的消息内容
     * @param session 当前会话对象
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        try {
            logger.info("收到客户端消息: fanId={},appId={}, message={}", this.fanId, this.appId, message);

            // 1. 心跳检测处理 (常见约定：客户端发送PING，服务器回复PONG)
            if ("PING".equalsIgnoreCase(message)) {
                session.getAsyncRemote().sendText("PONG");
            }

        } catch (Exception e) {
            logger.error("处理客户端消息失败: fanId={},appId={}", fanId, appId, e);
        }
    }

    @OnError
    public void onError(Session session, Throwable error) {
        logger.error("fanId[{}], appId[{}] 发生错误", fanId, appId, error);
    }

    public static int getOnlineClientSize() {
        return onlineClientSize.get();
    }

    /**
     * 发送消息
     * @param msg
     * @throws Exception
     */
    private void sendMessage(String msg) throws Exception {
        if (this.session.isOpen()) {
            this.session.getBasicRemote().sendText(msg);
        }
    }

    /**
     * 根据用户ID发送消息
     * @param msg
     * @throws Exception
     */
    private void sendMsgByFansIdAndAppId(String fanId,String appId, Object msg) {
        try {
            logger.info("推送ws消息到浏览器, fanId={},appId={}, msg={}", fanId,appId, msg);

            Set<MinProgramWsChannelServer> wsSet = wsUserIdMap.get(fanId);
            if (wsSet == null || wsSet.isEmpty()) {
                logger.info("fanId[{}], appId[{}] 无ws监听客户端", fanId, appId);
                return;
            }

            wsSet.stream()
                    .filter(item -> fanId == null ? true : fanId.equals(item.fanId))
                    .forEach(item -> {
                        try {
                            item.sendMessage(JSONUtil.toJsonStr(msg));
                        } catch (Exception e) {
                            logger.info("推送设备消息时异常，appId={}, fanId={}", appId, item.fanId, e);
                        }
                    });
        } catch (Exception e) {
            logger.info("推送消息时异常，appId={}", appId, e);
        }
    }

    /**
     *
     * @param fanId 用户ID
     * @param appId 小程序的AppId
     * @param eventDesc 事件描述 如: 导入商户数据
     * @param msg 发送的消息内容
     */
    public void convertAndSend(String fanId, String appId, String eventDesc, String msg) {
        WebSocketFansEvent event = new WebSocketFansEvent();
        event.setPayloadType(WebSocketConstants.PayloadType.SINGLE_ENTITY);
        event.setTopic(fanId + "/" + appId);
        event.setEventDesc(eventDesc);
        event.setContent(msg);
        sendMsgByFansIdAndAppId(fanId, appId, JSONUtil.toJsonStr(event));
    }

}